package cn.doitedu.demo7;

import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/8
 * @Desc: 学大数据，上多易教育
 * <p>
 * 静态画像条件：
 * age>30, job in ['程序员','工程师','公务员']
 * <p>
 * 动态画像统计条件：
 * A事件发生至少 3次
 * A事件 p0属性 最大值> 20
 * <p>
 * <p>
 * 触发条件：
 * X 事件
 * <p>
 * {
 * "rule_id":"rule-m03-006",
 * "static_profile_conditions":[
 * {
 * "tag_name":"age",
 * "compare_type":">",
 * "compare_value":30
 * },
 * {
 * "tag_name":"job",
 * "compare_type":"in",
 * "compare_value":["程序员","工程师","公务员"]
 * }
 * ],
 * "dynamic_profile_condition":{
 * "event_id":"A",
 * "event_min_count":3,
 * "event_property_name":"p0",
 * "property_min_max":20
 * },
 * "trigger_event":"X"
 * }
 **/

public class RuleModelCalculator_03 implements RuleCalculator {


    private String ruleId;

    JSONObject paramObject;

    Roaring64Bitmap preSelectBitmap;
    ValueState<Integer> countState;
    ValueState<Integer> propertyMaxState;

    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext, RuleMeta ruleMeta) throws IOException {

        this.paramObject = JSON.parseObject(ruleMeta.getRule_param_json());

        this.ruleId = paramObject.getString("rule_id");

        this.preSelectBitmap = ruleMeta.getPreSelectBitmap();

        this.countState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId + "-a_cnt", Integer.class));
        this.propertyMaxState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId + "-max", Integer.class));

        message = new JSONObject();
        message.put("rule_id",ruleId);

    }

    @Override
    public void calc(UserAction userAction, Collector<String> out) throws IOException {
        // 如果本行为的行为人，不属于本规则的 预圈选人群，则什么也不做，直接返回
        if(! preSelectBitmap.contains(userAction.getUser_id())) return;

        String eventId = userAction.getEvent_id();

        // 取动态统计条件的目标事件和目标属性
        JSONObject dynamicProfileCondition = paramObject.getJSONObject("dynamic_profile_condition");

        String targetEventId = dynamicProfileCondition.getString("event_id");
        String targetPropertyName = dynamicProfileCondition.getString("event_property_name");
        int eventMinCount = dynamicProfileCondition.getIntValue("event_min_count");
        int propertyMinMax = dynamicProfileCondition.getIntValue("property_min_max");

        if (eventId.equals(targetEventId)) {

            // 发生次数+1
            countState.update(countState.value() == null ? 1 : countState.value() + 1);

            // 属性值替换最大值
            Integer oldMax = propertyMaxState.value();
            oldMax = oldMax == null ? Integer.MIN_VALUE : oldMax;

            // 从行为事件中取到目标属性的值
            String s = userAction.getProperties().get(targetPropertyName);
            int actionPropertyValue = Integer.parseInt(s);

            // 如果本次事件的目标属性值 比  状态中记录的原来的最大值  还大，则替换掉状态中的值
            if (actionPropertyValue > oldMax) {
                propertyMaxState.update(actionPropertyValue);
            }
        }


        // 取出触发条件事件id
        String triggerEvent = paramObject.getString("trigger_event");
        if (triggerEvent.equals(eventId) && countState.value() != null
                && countState.value() > eventMinCount
                && propertyMaxState.value() != null
                && propertyMaxState.value() > propertyMinMax) {

            message.put("user_id",userAction.getUser_id());
            message.put("hit_time",userAction.getAction_time());

            out.collect(message.toJSONString());

        }


    }

    @Override
    public String ruleId() {
        return this.ruleId;
    }

    @Override
    public boolean isNeedProcessPastData() {
        return false;
    }


}
